package com.zhoufy.client;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.zhoufy.service.JSONSerializer;
import com.zhoufy.service.RpcDecoder;
import com.zhoufy.service.RpcEncoder;
import com.zhoufy.service.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created with IntelliJ IDEA by ChouFy on 2020-04-14.
 *
 * @author zhoufy
 */
public class RpcConsumer {


    private static AtomicInteger counter = new AtomicInteger();


    private static ExecutorService executor;

    private static UserClientHandle userClientHandle;


    private static int availableProcessors = Runtime.getRuntime().availableProcessors();

    static {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("demo-pool-%d").build();

        executor = new ThreadPoolExecutor(availableProcessors, availableProcessors, 2L, TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(64), namedThreadFactory, new ThreadPoolExecutor.CallerRunsPolicy());

    }


    public Object createProxy(Class<?> clazz, Object providerName) {
        // 借助jdk动态代理生成代理对象
        return Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[]{clazz}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

                System.out.println("invoke");
                // 调用 初始化netty 客户端
                System.out.println(userClientHandle == null);
                if (userClientHandle == null) {
                    initClient();
                }
                System.out.println("设置参数");
                // 设置参数
                userClientHandle.setPara(providerName);

                System.out.println("执行线程");
                return executor.submit(userClientHandle).get();
            }
        });

    }


    public static void initClient() throws InterruptedException {
        userClientHandle = new UserClientHandle();


        EventLoopGroup group = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(userClientHandle);

                    }
                });
        bootstrap.connect("localhost", 9095).sync();


    }


}
